package com.shujia.flink.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo3MysqlSInk {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //统计单词的数量
    val countDS: DataStream[(String, Int)] = linesDS
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)

    /**
     * 将统计的结果保存到数据库中
     *
     */

    countDS.addSink(new MysqlSInk)

    env.execute()
  }
}

/**
 * SinkFunction: 自定义sink
 * RichSinkFunction: 多个open和close方法，用于打开和关闭网络连接
 *
 */
class MysqlSInk extends RichSinkFunction[(String, Int)] {

  var con: Connection = _
  var stat: PreparedStatement = _

  /**
   * open方法每一个task中执行一次，在task初始化的时候执行
   *
   * @param parameters ： flink环境参数对象
   */
  override def open(parameters: Configuration): Unit = {
    println("task初始化，创建网络链接")
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //2、创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata", "root", "123456")
    //3、编写插入数据的sql
    //replace into ： 如果不存在插入，如果存在就更新
    stat = con.prepareStatement("replace into word_count(word,c) values(?,?)")

  }

  //在task回收的时候执行
  override def close(): Unit = {

    //关闭链接
    stat.close()
    con.close()
  }

  /**
   * invoke: 每一条数据执行一次
   *
   * @param kv      ：数据
   * @param context ：上下文对象
   */
  override def invoke(kv: (String, Int), context: SinkFunction.Context): Unit = {
    /**
     * 将数据保存到mysql中  - JDBC
     *
     */
    //4、指定sql的参数
    stat.setString(1, kv._1)
    stat.setLong(2, kv._2)
    //5、执行插入
    stat.execute()

  }
}
